package com.hulk.rpc.transport.client;

import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.serialize.Serializer;
import com.hulk.rpc.transport.client.future.RequestWithFuture;
import com.hulk.rpc.transport.client.handler.HulkChannelInitializer;
import com.hulk.rpc.transport.client.sender.BatchSender;
import com.hulk.rpc.transport.client.sender.Sender;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Objects;

/**
 * @author xuxukang
 * @date 2019-04-08 10:05
 */
@Slf4j
@Getter
final class NettyClient implements Closeable {

  private final Serializer serializer;
  private final HostPort serverAddress;
  private final EventLoopGroup eventLoopGroup;
  private final int connectCount;

  private volatile Sender[] senders;
  private volatile HostPort clientAddress;

  public NettyClient(Serializer serializer, HostPort serverAddress,
                     EventLoopGroup eventLoopGroup, int connectCount) {
    this.serializer = serializer;
    this.serverAddress = serverAddress;
    this.eventLoopGroup = eventLoopGroup;
    this.connectCount = connectCount;
  }

  @Override
  public void close() throws IOException {
    if (senders == null) {
      return;
    }

    final Sender[] senders = this.senders;
    this.senders = null;

    for (Sender sender : senders) {
      try {
        sender.close();
      } catch (Exception e) {
        if (log.isWarnEnabled()) {
          log.warn("关闭出错", e);
        }
      }
    }
  }

  int connectCount() {
    return connectCount;
  }

  /**
   * 发送rpc请求
   *
   * @param channelIndex  客户端channel下标(一个App可能有多个连接)
   * @param futureRequest 异步请求对象
   */
  void send(int channelIndex, RequestWithFuture futureRequest) {
    Objects.requireNonNull(futureRequest, "request with future can not be null");
    senders[channelIndex].send(futureRequest);
  }

  /**
   * 创建netty客户端(初始化sender)
   */
  void connect() throws InterruptedException {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(eventLoopGroup);

    bootstrap.option(ChannelOption.SO_REUSEADDR, true);
    bootstrap.option(ChannelOption.SO_RCVBUF, 256 * 1024);
    bootstrap.option(ChannelOption.SO_SNDBUF, 256 * 1024);
    bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, //
            new WriteBufferWaterMark(1024 * 1024, 2048 * 1024));

    if (eventLoopGroup instanceof EpollEventLoopGroup) {
      bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
      bootstrap.channel(EpollSocketChannel.class);
    } else if (eventLoopGroup instanceof NioEventLoopGroup) {
      bootstrap.channel(NioSocketChannel.class);
    }

    bootstrap.handler(new HulkChannelInitializer(serializer));

    Sender[] newSenders = new Sender[connectCount];
    for (int i = 0; i < connectCount; i++) {
      Channel channel = bootstrap.connect(serverAddress.getHost(), serverAddress.getPort()).sync().channel();
      newSenders[i] = new BatchSender(channel);

      if (log.isInfoEnabled()) {
        log.info(serverAddress + " connect " + i + "/" + connectCount);
      }

      if (i == 0) {
        InetSocketAddress insocket = (InetSocketAddress) channel.localAddress();
        clientAddress = new HostPort(insocket.getAddress().getHostAddress(), 0);
      }
    }

    Sender[] old = senders;
    senders = newSenders;

    if (old != null) {
      for (Sender sender : old) {
        try {
          sender.close();
        } catch (Exception e) {
          if (log.isWarnEnabled()) {
            log.warn("关闭出错", e);
          }
        }
      }
    }
  }
}
